package com.vtron.mq.service.impl;

import com.alibaba.fastjson.JSON;
import com.vtron.mq.models.MsgPackage;
import com.vtron.mq.models.PackageType;
import com.vtron.mq.service.IMsgPackageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @date : 2022/4/26 16:39
 * @author: 忻龙彪
 */
@Component
public class MsgPackageServiceImpl implements IMsgPackageService {

    private final Logger logger = LoggerFactory.getLogger(MsgPackageServiceImpl.class);

    /**
     * 消息主题
     */
 /*   @Value("${kafka.topic}")*/
    private String kafkaTopic="test";
    /**
     * kafka服务模板
     */
    @Autowired(required = false)
    private  KafkaTemplate<String, String> kafkaTemplate;


    @Override
    public void send(MsgPackage msg) {
        String msgBody=JSON.toJSONString(msg);
        kafkaTemplate.send(kafkaTopic, msgBody).addCallback(r->{
            logger.info("kafka send sucess=>"+msgBody);
        },e->{
            logger.error("kafka send failure=>"+msgBody+",error=>"+e.getMessage());
        });
    }

    @Override
    public void send(MsgPackage msg, ListenableFutureCallback<SendResult<String, String>> callback) {
        kafkaTemplate.send(kafkaTopic, JSON.toJSONString(msg)).addCallback(callback);
    }

    @Override
    public void send(String topic, MsgPackage msg) {
        String msgBody=JSON.toJSONString(msg);
        kafkaTemplate.send(topic, msgBody).addCallback(r->{
            logger.info("kafka send sucess=>"+msgBody);
        },e->{
            logger.error("kafka send failure=>"+msgBody+",error=>"+e.getMessage());
        });
    }

    @Override
    public void send(String topic, MsgPackage msg, ListenableFutureCallback<SendResult<String, String>> callback) {
        kafkaTemplate.send(topic, JSON.toJSONString(msg)).addCallback(callback);
    }

    @Override
    public void send(PackageType type, String body) {
        String msgBody=JSON.toJSONString(new MsgPackage(type,body));
        kafkaTemplate.send(kafkaTopic, msgBody).addCallback(r->{
            logger.info("kafka send sucess=>"+msgBody);
        },e->{
            logger.error("kafka send failure=>"+msgBody+",error=>"+e.getMessage());
        });
    }

    @Override
    public void send(PackageType type, String body, ListenableFutureCallback<SendResult<String, String>> callback) {
        kafkaTemplate.send(kafkaTopic, JSON.toJSONString(new MsgPackage(type,body))).addCallback(callback);
    }

    @Override
    public void send(String topic, PackageType type, String body) {
        String msgBody=JSON.toJSONString(new MsgPackage(type,body));
        kafkaTemplate.send(topic, msgBody).addCallback(r->{
            logger.info("kafka send sucess=>"+msgBody);
        },e->{
            logger.error("kafka send failure=>"+msgBody+",error=>"+e.getMessage());
        });
    }

    @Override
    public void send(String topic, PackageType type, String body, ListenableFutureCallback<SendResult<String, String>> callback) {
        kafkaTemplate.send(topic, JSON.toJSONString(new MsgPackage(type,body))).addCallback(callback);
    }
}
